package org.renhj.metric;

import com.alibaba.fastjson.JSONObject;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.renhj.config.ZookeeperConfig;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;

public class ZookeeperMetric extends TimerTask {
    public static volatile ZookeeperMetric zookeeperMetric;

    private final CuratorFramework curatorFramework;
    private final static ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);

    private final static String ZOOKEEPER_HOST = ZookeeperConfig.ZOOKEEPER_HOST;
    private final static int ZOOKEEPER_PORT = ZookeeperConfig.ZOOKEEPER_PORT;
    private final static String METRIC_PATH = ZookeeperConfig.METRIC_PATH;


    private final static Map<String, String> REQUEST_ADDRESS_MAP = new ConcurrentHashMap<>();
    private final static Map<String, Metric> REQUEST_METRIC_MAP = new ConcurrentHashMap<>();

    private final static Map<String, Metric> ADDRESS_METRIC_MAP = new ConcurrentHashMap<>();

    // 私有化构造方法
    private ZookeeperMetric(String host, int port) throws Exception {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(5000, 3);
        curatorFramework = CuratorFrameworkFactory.builder()
                .retryPolicy(retryPolicy)
                .connectionTimeoutMs(5000)
                .connectString(host + ":" + port)
                .namespace("curator")
                .build();
        curatorFramework.start();

        // 设置定时任务，每隔5秒将监控数据上报到zookeeper
        executor.scheduleAtFixedRate(this, 0, 5, TimeUnit.SECONDS);
    }

    // 检查并创建zookeeper监控节点 /curator/metric/127.0.0.1:9000  {address:127.0.0.1:9000, start: 19123123132, const: 34ms}
    public void createMetricNode(String requestId, String address) throws Exception {
        REQUEST_ADDRESS_MAP.put(requestId, address);
        REQUEST_METRIC_MAP.put(requestId, new Metric(address, System.currentTimeMillis(), null, 0L));
    }

    public static ZookeeperMetric getInstance() throws Exception {
        if (zookeeperMetric == null) {
            synchronized (ZookeeperMetric.class) {
                zookeeperMetric = new ZookeeperMetric(ZOOKEEPER_HOST, ZOOKEEPER_PORT);
            }
        }
        return zookeeperMetric;
    }

    @Override
    public void run() {
        // 将监控数据上报

        /*
            1、判断是否存在节点 /curator/metric
            2、不存在创建节点
            3、存在的话，创建临时节点 /curator/metric/127.0.0.1:9000，并将监控数据写入到临时节点数据中
        */
        try {
            Stat stat = curatorFramework.checkExists().forPath(METRIC_PATH);
            if (stat == null) {
                curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(METRIC_PATH);
            }
            Set<Map.Entry<String, Metric>> entries = ADDRESS_METRIC_MAP.entrySet();
            for (Map.Entry<String, Metric> entry : entries) {
                // 判断metric最后一次请求时间距离现在是否超过5s，不超过的话，上报，否则删除zookeeper上的节点
                Metric metric = entry.getValue();
                String path = METRIC_PATH + "/" + entry.getKey();

                if (System.currentTimeMillis() - metric.getStart() < 5000) {
                    Stat stat1 = curatorFramework.checkExists().forPath(path);
                    if (stat1 == null) {
                        curatorFramework.create().withMode(CreateMode.EPHEMERAL).forPath(path, metric.getCost().toString().getBytes());
                    }
                    curatorFramework.setData().forPath(path, metric.getCost().toString().getBytes());
                } else {
                    curatorFramework.delete().forPath(METRIC_PATH + "/" + entry.getKey());
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public void updateMetric(String requestId, long currentTimeMillis) {
        Metric metric = REQUEST_METRIC_MAP.get(requestId);
        metric.setEnd(currentTimeMillis);
        metric.calcCost();

        ADDRESS_METRIC_MAP.put(REQUEST_ADDRESS_MAP.get(requestId), metric);

        REQUEST_METRIC_MAP.remove(requestId);
        REQUEST_ADDRESS_MAP.remove(requestId);
    }

    public Map<String, Metric> getMetrics() throws Exception {
        return ADDRESS_METRIC_MAP;
    }
}
